package com.ztao.kafka.controller;

import lombok.SneakyThrows;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Date;
import java.util.concurrent.ExecutionException;

@RestController
public class KafkaController {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
//    @GetMapping("/send")
//    public String send() {
//        kafkaTemplate.send("topic1","要发送的数据"+new Date().toLocaleString());
//        return "发送消息success";
//    }

//同步执行
//    @RequestMapping("/send")
//    public String sendMsg() throws ExecutionException, InterruptedException {
//
//        System.out.println("开始发送");
//        //send的时候能不知道发送失败了，应不应该知道发送失败
//        //如果发送失败，消息再次处理
//        //这个send方法本身就是异步处理的
//        // kafkaTemplate.send("topic1", "要发送的数据" + new Date().toLocaleString());
//
//        //可以接收到这个方法的返回值，已返回值来判断是否发送成功
//
//        //.get（）接收到返回值 ，并且使用了get方法，这个方法就变成了同步执行；
//        SendResult<String, String> topic1 = kafkaTemplate.send("topic1", "要发送的数据" + new Date().toLocaleString())
//                .get();
//        System.out.println("发送的返回值"+topic1);
//
//        System.out.println("发送结束");
//        return "发送消息";
//    }



    //异步发送
    @SneakyThrows
    @RequestMapping("/send")
    public String sendMsg() {

        System.out.println("开始发送");
        //send的时候能不呢知道发送失败了，应不应该知道发送失败
        //如果发送失败，消息再次处理
        //这个send方法本身就是异步处理的
        // kafkaTemplate.send("topic1", "要发送的数据" + new Date().toLocaleString());

        //可以接收到这个方法的返回值，已返回值来判断是否发送成功

        //.get（）接收到返回值 ，并且使用了get方法，这个方法就变成了同步执行；
   /* SendResult<String, String> topic1 = kafkaTemplate.send("topic1", "要发送的数据" + new Date().toLocaleString())
            .get();
    System.out.println("发送的返回值"+topic1);

   */
        //异步处理

        kafkaTemplate.send("topic1", "要发送的数据" + new Date().toLocaleString())
                .addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                    @Override
                    public void onFailure(Throwable throwable) {
                        System.out.println("消息发送失败");

                        //可以把失败的信息给收集起来  --logs
                    }

                    @Override
                    public void onSuccess(SendResult<String, String> stringStringSendResult) {
                        System.out.println("消息发送成功");
                    }
                });

        System.out.println("发送结束");
        return "发送消息";
    }


}
